package tss.layer;

import java.io.*;
import java.util.*;

import tss.*;
import tss.util.*;

/**
 * <p>Cbcast implements a causally ordered multicast mechanism on top of
 * a reliable multicast. 
 * 
 * <p>'Reliable' means that multicast messages are delivered best-effort
 * but that they may appear in any order
 * 
 * <p>'Causally ordered' means ordered with respect to Lamport's "happens-before"
 * relation, '->', so that:
 * 
 * <p><pre>
 *   For all messages m1,m2: send(m1) -> send(m2) => deliver(m1) -> deliver(m2)
 * </pre>
 * 
 * <p>In the context of this class, this means that if m2 is received before
 * m1 (in local time), then m2 will be queued until m1 is received and following
 * this, m1 will be delivered then m2 will be delivered)
 * 
 * <p>The following additional properties are required:
 * 
 * <p>"group.size" : a positive integer denoting the number of members of the
 *                (static) group that this process is a member of
 * 
 * @author Klaus Marius Hansen, klaus.m.hansen@daimi.au.dk
 * 
 */
public class CbcastLayer extends Layer {
	// The vector timestamp that this transport handles
	public VectorTimestamp timestamp;
	// A set of pending messages that cannot yet be delivered
	private Collection<Message> buffer;
	
	public void setProperties(String ... properties) {
		super.setProperties(properties);
		timestamp = new VectorTimestampImpl(new Integer(getProperty("group.size")), id);
		buffer = new LinkedList<Message>();
	}
	
	public void down(Message message) throws IOException {
		timestamp.increment();
		message.pushHeader(timestamp);
		super.down(message);
	}

	public void up(Message message) {
		buffer.add(message);
		while ((message = removeDeliverable()) != null) {
			timestamp.max((VectorTimestamp) message.popHeader());
			super.up(message);
		}
	}
	
	/**
	 * Check whether any messages in buffer are deliverable (compared to our
	 * current timestamp), remove such a message, and return it
	 * 
	 * @return a deliverable message or null if no message is deliverable
	 */
	private Message removeDeliverable() {
		Message result = null;
		for (Message message : buffer) {
			if (timestamp.canDeliver((VectorTimestamp) message.peekHeader())) {
				buffer.remove(message);
				result = message;
				break;
			}
		}
		return result;
	}
}
